package db;

import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Response;
import tlsb.DB;
import tlsb.DBException;
import tlsb.Utils;
import util.Pair;
import util.RedisCluster;
import util.RedisClusterFactory;

public class RedisClientOVLP extends DB {

	final private long STARTOFFSET = 0;
	final private long ENDOFFSET = 19;
	final private int MAX = 999999999;
	final public static String SETOFOWNEDTL = ":own";
	final public static String SETOFBELONGEDTL = ":blg";
	final private String FSTCOMFIELD = "fstc";
	final private int HTLLENGTH = 20;
	final private String PREFIXOFCIDFORSID = "cfs";
	// final private String UIDOFMID = ""

	static AtomicLong midCounter = new AtomicLong(0);

	static Random random = new Random();

	private RedisCluster folRedisCluster = RedisClusterFactory.instance
			.getFolRedisCluster();
	private RedisCluster statusRedisCluster = RedisClusterFactory.instance
			.getStatusRedisCluster();
	private RedisCluster metaDataRedisCluster = RedisClusterFactory.instance
			.getMetaDataRedisCluster();

	private boolean _Debug = false;
	private final int OperationRetries = 10;

	public void init() throws DBException {

	}

	public void cleanup() throws DBException {
		folRedisCluster.close();
		statusRedisCluster.close();
		metaDataRedisCluster.close();
	}

	@Override
	public int follow(String uid, String fid) {
		// TODO Auto-generated method stub
		// add to new followers and followers for different strateges; also
		// update owns(first) and belongs
		return 0;
	}

	@Override
	// TODO uid change string into integer.
	public int getHTL(String uid, int start) {

		// /String cid = "";
		// if ((cid = jedis.hget(uid, FSTCOMFIELD)) == null) {
		// TODO 是否保存在本地
		// if ((cid = Utils.getCidOfUid(uid)) == null) {
		// System.err.println(uid + " has no cccccccid...");
		// return 0;

		// }
		Set<String> sids = folRedisCluster.getJedis(
				folRedisCluster.getPartition().shard(uid)).smembers(
				Utils.joinStr(uid, SETOFOWNEDTL));// get
		// all
		// stls
		// the
		// user shares

		if (sids == null) {
			// System.err.println(uid + " has no stls...");
			return 0;
		}
		List<List<String>> stls = new LinkedList<List<String>>();
		List<Response<List<String>>> responses = new LinkedList<Response<List<String>>>();
		for (String sid : sids) {
			/*
			 * int nid =
			 * CGA.hashByCid(jedis.get(Utils.joinStr(PREFIXOFCIDFORSID, sid)));
			 */
			int nid = statusRedisCluster.getPartition().shard(sid);
			responses.add(statusRedisCluster.getPipeline(nid).lrange(sid,
					STARTOFFSET, ENDOFFSET));
		}
		statusRedisCluster.synAllPipelines();
		for (Response<List<String>> response : responses) {
			stls.add(response.get());
		}
		// merge stls(removing duplicates and filter newly unfollows)
		String[] htl = mergeFilter(stls, HTLLENGTH, start);
		if (_Debug) {
			System.out.println(uid + " OVLP:" + htl);
		}
		return 0;
		// }
		// return 1;
	}

	@Override
	public int newUser() {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	public int post(String uid, String mid) {
		/*
		 * String cid = ""; // if ((cid = jedis.hget(uid, FSTCOMFIELD)) == null)
		 * { if ((cid = Utils.getCidOfUid(uid)) == null) {
		 * System.err.println(uid + " has no cccccccid...post"); return 0; }
		 */
		Jedis thisJedis = folRedisCluster.getJedis(folRedisCluster
				.getPartition().shard(uid));
		Set<String> sids = thisJedis.smembers(Utils.joinStr(uid,
				SETOFBELONGEDTL));
		if (sids == null) {
			System.err.println(uid + " has no stls to push");
			return 0;
		}

		for (String sid : sids) {
			int nid = statusRedisCluster.getPartition().shard(sid);
			// statusRedisCluster.getPartition().shard(
			// Utils.getCidOfSid(sid));

			statusRedisCluster.getPipeline(nid).lpush(sid, mid);
			midCounter.incrementAndGet();
			if (midCounter.get() % 1000000 == 0)
				System.out.println(midCounter.get());
		}
		statusRedisCluster.synAllPipelines();
		return 0;
		// }
		// return 1;
	}

	@Override
	public int unfollow(String uid, String fid) {
		// TODO Auto-generated method stub
		return 0;
	}

	private static class MidQueueComparator implements
			Comparator<Pair<String, ListIterator<String>>> {

		public static MidQueueComparator instance = new MidQueueComparator();

		@Override
		public int compare(Pair<String, ListIterator<String>> arg0,
				Pair<String, ListIterator<String>> arg1) {
			return 0- arg0.a.compareTo(arg1.a);
		}

	}

	public static String[] mergeFilter(List<List<String>> stls, int count, int start) {
		String[] htl = new String[count];		
		String lastMid = "-1";

		/**
		 * init
		 */
		PriorityQueue<Pair<String, ListIterator<String>>> midQueue = new PriorityQueue<Pair<String, ListIterator<String>>>(
				100, MidQueueComparator.instance);
		for (List<String> stl : stls) {
			ListIterator<String> iter = stl.listIterator(stl.size());
			if (iter.hasPrevious()) {
				midQueue.add(new Pair<String, ListIterator<String>>(iter.previous(),
						iter));
			}
		}
		
		int htlCount = 0;
		
		while (htlCount < start + count && !midQueue.isEmpty()) {
			Pair<String, ListIterator<String>> pair = midQueue.poll();
			if (pair == null)
				break;
			
			if (!lastMid.equals(pair.a)) {
				if (htlCount >= start) {
					htl[htlCount - start] = pair.a;
				}
				htlCount++;
			}
			if (pair.b.hasPrevious()) {
				pair.a = pair.b.previous();
				midQueue.add(pair);
			}
		}
		
		/*for (int i = 0; i < count + start; i++) {
			
			for (int j = 0; j < stls.size(); j++) {
				// check
				thisStl = stls.get(j);
				if (thisStl.size() == 0)
					continue;
				thisMid = thisStl.get(thisStl.size() - 1);
				// remove duplicates
				if (thisMid.equals(lastMid)) {
					thisStl.remove(thisStl.size() - 1);
					if (thisStl.size() == 0)
						continue;
					thisMid = thisStl.get(thisStl.size() - 1);
				}

				if (nextMid == null
						|| Long.parseLong(nextMid) < Long.parseLong(thisMid)) {
					nextMid = thisMid;
					stlNo = j;
				}
			}
			if (stlNo == MAX)
				break;
			if (i >= start) {
				htl[i] = nextMid;
			}
			stls.get(stlNo).remove(stls.get(stlNo).size() - 1);
			lastMid = nextMid;
			nextMid = null;
			stlNo = MAX;
		}
*/
		return htl;
	}
}
